package mya.mes.mdc.mqtt;


import mya.mes.mdc.share.mqttread.MqttMsgProcessor;
import mya.mes.mdc.share.vo.MdcOptions;
import mya.mes.mdc.share.vo.mqtt.MqttMsgReceived;
import mya.mes.mdc.share.vo.mqtt.MsgProcessTaskResult;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class MsgReceiver extends Thread {

    private final ExecutorService executorService;
    private final MqttReceivedQueue mqttMsgQueue;

    private MdcOptions mdcOptions;

    public MsgReceiver(MdcOptions mdcOptions, MqttReceivedQueue mqttMsgQueue) {
        this.mdcOptions = mdcOptions;
        executorService = Executors.newSingleThreadExecutor();
        this.mqttMsgQueue = mqttMsgQueue;
        this.setDaemon(true);
    }

    @Override
    public void run() {
        while (true) {
            try {
                MqttMsgReceived mqttPayload = mqttMsgQueue.take();
                Future<MsgProcessTaskResult> future = processMsg(mqttPayload);
//                System.out.println("MsgReceiver run:"+currentThread().getName()+"mqttMsgQueue_hashcode"+mqttMsgQueue.hashCode());
                try {
                    MsgProcessTaskResult result = future.get();
                    if (!result.success) {
                        //记录错误
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }

            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }

    public Future<MsgProcessTaskResult> processMsg(MqttMsgReceived mqttPayload) {
        MqttMsgProcessor processor = mdcOptions.MQTT_RECEIVE_PROCESSOR;
        if (processor == null) {
            return null;
        }
        MsgProcessTask task = new MsgProcessTask(mqttPayload, processor);
        return executorService.submit(task);
    }


    @Override
    protected void finalize() throws Throwable {
        if (executorService != null) {
            executorService.shutdown();
        }
        super.finalize();
    }
}
